package com.epic;

import java.util.Properties;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaConsumerUtils {
    public static KafkaConsumer getConsumer(String kafkaHost, String groupId) {
        return getConsumer(kafkaHost, groupId, true);
    }

    public static KafkaConsumer getConsumer(String kafkaHost, String groupId, boolean isAutoOffset) {
        Properties props = new Properties();
        props.put("bootstrap.servers", kafkaHost);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", isAutoOffset ? "true" : "false");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RangeAssignor");
        KafkaConsumer<String, String> consumer = new KafkaConsumer(props);
        return consumer;
    }
}
